# +----------------------------------------------------------------------
# | ChatWork智能聊天办公系统
# +----------------------------------------------------------------------
# | 软件声明: 本系统并非自由软件,未经授权任何形式的商业使用均属非法。
# | 版权保护: 任何企业和个人不允许对程序代码以任何形式任何目的复制/分发。
# | 授权要求: 如有商业使用需求,请务必先与版权所有者取得联系并获得正式授权。
# +----------------------------------------------------------------------
# | Author: ChatWork Team <2474369941@qq.com>
# +----------------------------------------------------------------------
import copy
import json
import time
import os.path
import asyncio
from decimal import Decimal
from typing import List, Dict
from fastapi import Request
from tortoise.models import in_transaction
from exception import AppException
from common.utils.urls import UrlUtil
from common.utils.files import FilesUtil
from common.utils.tools import ToolsUtil
from common.utils.config import ConfigUtil
from common.enums.agent import AgentEnum
from common.enums.wallet import WalletEnum
from common.models.users import UserModel
from common.models.users import UserWalletModel
from common.models.dev import DevModelsModel
from common.models.agents import AgentMcpModel
from common.models.agents import AgentKnowModel
from common.models.agents import AgentRobotModel
from common.models.agents import AgentSquareModel
from common.models.agents import AgentAttachmentModel
from common.models.agents import AgentChatRecordsModel
from common.models.agents import AgentRobotTeamedModel
from common.models.agents import AgentRobotPublishModel
from common.models.agents import AgentRobotInstructModel
from common.cache.pool_cache import KeyPoolCache
from common.chain.mcp_server import MCPClientPool
from common.chain.chain_server import FlowsSchema, ChatUtils, AIChannelFactory, AIWebSearch
from common.service.retrieval_logic import RecallKnowLogic, RecallUtil
from plugins.censor.driver import CensorDriver


class ChatRobotWidget:

    def __init__(self, request: Request, post: dict):
        # 对话参数
        self.start_time = time.time()
        self.request = request
        self.stream: bool = bool(post.get('stream', False))
        self.search: bool = bool(post.get('search', False))
        self.ip: str = str(post.get("ip"))
        self.sid: int = int(post.get("sid"))
        self.bot: str = str(post.get("bot"))
        self.scene: str = str(post.get("scene"))
        self.user_id: int = int(post.get("user_id"))
        self.publish_id: int = int(post.get("publish_id"))
        self.publish_uid: str = str(post.get("publish_uid"))

        # 输入数据
        self.user_inputs: list = post.get("messages")
        self.question: str = ""  # 用户的问题
        self.attach: list = []   # 提问附带件

        # 引用数据
        self.document: str = ""   # 知识库内容
        self.adjuncts: list = []  # 知识库附件
        self.messages: list = []  # 消息记录组
        self.contexts: list = []  # 记录上下文
        self.quotes: list = []    # 引用的知识

        # 记录ID
        self.record_id: int = 0

        # MCP服务
        self.mcp_servers: dict = {}
        self.mcp_tools: dict = {}

        # 应用信息
        self.know_ids: List[int] = []
        self.robot = None
        self.user = None

        # 对话模型
        self.chat_mark: str = ""
        self.chat_channel: str = ""
        self.chat_alias: str = ""
        self.chat_model: str = ""
        self.chat_price: int = 0
        self.chat_config: Dict[str, any] = {}

        # 向量模型
        self.vector_mark: str = ""
        self.vector_alias: str = ""
        self.vector_model: str = ""
        self.vector_price: int = 0

        # 重排模型
        self.ranking_mark: str = ""
        self.ranking_price: int = 0

        # 联网搜索
        self.search_status: bool = False
        self.search_engine: str = ""
        self.search_apikey: str = ""
        self.search_points: int = 0

        # 响应数据
        self.think: str = ""
        self.reply: str = ""
        self.annex: List[dict] = []
        self.tools_list: List[dict] = []
        self.web_search: List[dict] = []
        self.flows_usage: List[FlowsSchema] = []

    async def chat_sse(self):
        """ SSE请求 """
        try:
            # 初始数据
            start_time = time.time()
            await self._initialize()

            # 查知识库
            learning = await self._call_learning()

            # 查询指令
            instruct = await self._get_instruct()

            # 回复处理
            usages: dict = {}
            cancel: bool = False
            idx: str = str(time.time())
            if instruct:
                # 指令回复
                self.annex = instruct["annex"]
                self.reply = instruct["content"]
                yield ChatUtils.output_success(idx=idx, event="chat", data=self.reply, model=self.chat_alias)
                for ax in instruct["annex"]:
                    ev = ax["file_type"]
                    dt = json.dumps(ax, ensure_ascii=False)
                    yield ChatUtils.output_success(idx=idx, event=ev, data=dt, model=self.chat_alias)
            elif self.robot.empty_search_type == "default" and not learning:
                # 空值回复
                self.reply = self.robot.empty_search_text
                yield ChatUtils.output_success(idx=idx, event="chat", data=self.reply, model=self.chat_alias)
            elif self.robot.is_original_reply and learning:
                # 原文回复
                for item in self.quotes:
                    data = "\n\n" + item["answer"] if self.reply else item["answer"]
                    self.reply += data
                    yield ChatUtils.output_success(idx=idx, event="chat", data=data, model=self.chat_alias)
            else:
                # AI回复
                apikey = await self._get_apikey()
                aiServer = AIChannelFactory.create(self.chat_channel, self.chat_config, apikey)
                try:
                    # 联网检索
                    if self.search:
                        yield ChatUtils.output_success(event="search")
                        await self._web_searching()
                        json_data_output = json.dumps(self.web_search, ensure_ascii=False)
                        yield ChatUtils.output_success(event="search", data=json_data_output)

                    # 取上下文
                    messages = await self._get_context()

                    # 服务连接
                    if self.mcp_servers:
                        mcp_client = MCPClientPool()
                        await mcp_client.connect_to_servers(self.mcp_servers)
                        stream = mcp_client.chat_loop(aiServer, messages)
                    else:
                        stream = await aiServer.llm(messages)

                    # 流式提问
                    async for chunk in stream:
                        if not chunk.choices:
                            raise Exception(f"{chunk.code}: {chunk.message}")

                        think = ""
                        event = "chat"
                        idx = chunk.id
                        delta = chunk.choices[0].delta.model_dump()
                        tool_calls = None

                        if chunk.choices[0].finish_reason == "tool_calls":
                            resp = json.loads(chunk.choices[0].delta.content or "{}")
                            chunk.choices[0].delta.content = None
                            delta["content"] = None
                            event = "tools"
                            tool_calls = []
                            for tool in chunk.choices[0].delta.tool_calls:
                                tool_code, tool_name = tool.function.name.split("_", 1)
                                mcp = self.mcp_tools.get(tool_code) or {}
                                cal = {
                                    "id": tool.id,
                                    "code": tool_code,
                                    "name": mcp.get("name") + " / " + tool_name,
                                    "tool_name": tool_name,
                                    "tool_logo": mcp.get("logo"),
                                    "arguments": tool.function.arguments,
                                    "response": resp.get(tool.function.name, "")
                                }
                                tool_calls.append(cal)
                                exist_item = next((item for item in self.tools_list if item["id"] == cal["id"]), None)
                                if exist_item:
                                    exist_item.update(cal)
                                else:
                                    self.tools_list.append(cal)

                        if chunk.choices[0].finish_reason == "normal":
                            event = "finish"
                        if delta.get("reasoning_content"):
                            think = chunk.choices[0].delta.reasoning_content
                            self.think += chunk.choices[0].delta.reasoning_content
                        if delta.get("content"):
                            self.reply += chunk.choices[0].delta.content or ""
                        if chunk.usage:
                            usages = {
                                "total_tokens": chunk.usage.total_tokens,
                                "prompt_tokens": chunk.usage.prompt_tokens,
                                "completion_tokens": chunk.usage.completion_tokens
                            }

                        yield ChatUtils.output_success(
                            idx=idx,
                            event=event,
                            model=self.chat_model,
                            index=chunk.choices[0].index,
                            data=chunk.choices[0].delta.content,
                            think=think,
                            finish_reason=chunk.choices[0].finish_reason,
                            refusal=chunk.choices[0].delta.refusal,
                            tool_calls=tool_calls,
                            function_call=chunk.choices[0].delta.function_call
                        )

                    # 附件输出
                    self.annex = self.adjuncts
                    for att in self.annex:
                        ev: str = att["file_type"]
                        dt: str = json.dumps(att, ensure_ascii=False)
                        yield ChatUtils.output_success(idx=idx, event=ev, data=dt, model=self.chat_alias)
                except asyncio.CancelledError:
                    cancel = True

            # 上下文为空时
            if not self.messages:
                await self._get_context()

            # 对话使用计算
            if not usages:
                usages = ChatUtils.usage(self.messages, self.reply)

            # 对话消耗情况
            if usages:
                consume_points = ChatUtils.compute_price(usages["total_tokens"], self.chat_price)
                self.flows_usage.append(FlowsSchema(
                    name="AI 对话",
                    scene="chat",
                    alias=self.chat_alias,
                    model=self.chat_model,
                    task_time=f"{(time.time() - start_time):.2f}",
                    use_points=consume_points,
                    total_tokens=usages["total_tokens"],
                    prompt_tokens=usages["prompt_tokens"],
                    completion_tokens=usages["completion_tokens"],
                ))

            # 客户端主动取消
            if cancel:
                task = asyncio.create_task(self._save_record())
                await asyncio.shield(task)
            else:
                # 保存记录
                await self._save_record()
                # 相关问题
                if self.robot.question_guide:
                    yield ChatUtils.output_success(idx=idx, event="related", model=self.chat_alias)
                    await self._make_related()
                yield ChatUtils.output_success(idx=idx, model=self.chat_alias)
                yield "data: [DONE]"
        except AppException as e:
            yield ChatUtils.output_error(model=self.chat_model, code=e.code, message=e.msg)
        except Exception as e:
            yield ChatUtils.output_error(model=self.chat_model, message=str(e))

    async def chat_http(self):
        """ HTTP请求 """
        try:
            # 初始数据
            start_time = time.time()
            await self._initialize()

            # 查知识库
            learning = await self._call_learning()

            # 查询指令
            instruct = await self._get_instruct()

            # 回复处理
            usages: dict = {}
            if instruct:
                # 指令回复 material=材料
                self.annex = instruct["annex"]
                self.reply = instruct["content"]
            elif self.robot.empty_search_type == "default" and not learning:
                # 空值回复
                self.reply = self.robot.empty_search_text
            elif self.robot.is_original_reply and learning:
                # 原文回复
                for item in self.quotes:
                    data = "\n\n" + item["answer"] if self.reply else item["answer"]
                    self.reply += data
            else:
                # 联网检索
                await self._web_searching()
                # 取上下文
                messages = await self._get_context()
                # 发起回复
                apikey = await self._get_apikey()
                aiServer = AIChannelFactory.create(self.chat_channel, self.chat_config, apikey)
                results = await aiServer.llm(messages, stream=False)
                self.reply = results.choices[0].message.content
                self.attach = self.adjuncts
                if results.usage:
                    usages = {
                        "total_string": len(self.reply),
                        "total_tokens": results.usage.total_tokens,
                        "prompt_tokens": results.usage.prompt_tokens,
                        "completion_tokens": results.usage.completion_tokens
                    }

            # 对话使用情况
            if not usages:
                usages = ChatUtils.usage(self.messages, self.reply)

            # 对话消耗情况
            if usages:
                consume_points = ChatUtils.compute_price(usages["total_tokens"], self.chat_price)
                self.flows_usage.append(FlowsSchema(
                    name=str("AI 对话"),
                    scene="chat",
                    alias=self.chat_alias,
                    model=self.chat_model,
                    task_time=f"{(time.time() - start_time):.2f}",
                    use_points=consume_points,
                    total_tokens=usages["total_tokens"],
                    prompt_tokens=usages["prompt_tokens"],
                    completion_tokens=usages["completion_tokens"],
                ))

            # 对话记录保存
            await self._save_record()

            # 返回请求结果

            return ChatUtils.output_success(
                data=self.reply,
                model=self.chat_alias,
                annex=self.annex,
                stream=False,
                usage={
                    "total_tokens": usages["total_tokens"],
                    "prompt_tokens": usages["prompt_tokens"],
                    "completion_tokens": usages["completion_tokens"]
                },

            )
        except AppException as e:
            raise AppException(msg=e.msg, code=e.code)
        except Exception as e:
            raise AppException(msg=str(e), code=1)

    async def _initialize(self):
        """ 初始数据 """
        # 获取用户问题
        inputs = self.user_inputs[len(self.user_inputs) - 1]
        if not isinstance(inputs["content"], list):
            self.question = inputs["content"]
        else:
            self.question = next((ctx["text"] for ctx in inputs["content"] if ctx.get("type") == "text"), None)

        # 查智能体
        robot = await AgentRobotModel.filter(code=self.bot).first()
        self.robot = robot
        if robot is None:
            raise AppException("智能体不存在,请刷新")
        if robot.is_delete:
            raise AppException("智能体已被拥有者下架")
        if not robot.llm_models:
            raise AppException("智能体请关联对话模型")

        if self.publish_id:
            self.user_id = self.robot.user_id
        elif self.robot.user_id != self.user_id:
            # 判断是不是参与者
            teamed = await AgentRobotTeamedModel.filter(user_id=self.user_id, robot_id=self.robot.id).first()

            # 判断是不是在广场
            if not teamed:
                square = await (AgentSquareModel
                                .filter(robot_id=self.robot.id)
                                .filter(is_enable=1)
                                .filter(audit_status=1)
                                .first())
                if not square:
                    raise AppException("您无使用权限")
                if square.is_delete:
                    raise AppException("智能体已下架")

        # 查知识库
        if robot.know_ids:
            k_ids = robot.know_ids.split(",")
            knows = await AgentKnowModel.filter(id__in=k_ids, is_delete=0).all()
            if knows:
                self.know_ids = [int(know.id) for know in knows]
                vector_model = await DevModelsModel.check_models(knows[0].vector_model, "vector")
                self.vector_mark = vector_model["mark"]
                self.vector_alias = vector_model["alias"]
                self.vector_model = vector_model["model"]
                self.vector_price = vector_model["price"]

        # 对话模型
        chat_model = await DevModelsModel.check_models(robot.llm_models, "chat")
        self.chat_mark = chat_model["mark"]
        self.chat_channel = chat_model["channel"]
        self.chat_alias = chat_model["alias"]
        self.chat_model = chat_model["model"]
        self.chat_price = int(chat_model["price"])
        self.chat_config = chat_model["config"]

        # 重排模型
        if self.robot.ranking_status:
            rank_model = await DevModelsModel.check_models(robot.ranking_model, "ranking")
            self.ranking_price = int(rank_model["price"])

        # 联网搜索
        if self.search:
            web_config = await ConfigUtil.get("web_search") or {}
            self.search_status = bool(web_config.get("status") or False)
            self.search_engine = str(web_config.get("engine") or "")
            self.search_apikey = str(web_config.get(self.search_engine, {}).get("api_key", ""))
            self.search_points = int(web_config.get(self.search_engine, {}).get("points", 0))

        # 查询MCP
        if robot.mcp_ids:
            mcp_srv = {}
            m_ids = robot.mcp_ids.split(",")
            mcp_list = await AgentMcpModel.filter(id__in=m_ids, is_delete=0).all()
            for mcp in mcp_list:
                tool_list = json.loads(mcp.tools or "[]")
                tool_list = [tool for tool in tool_list if tool.get("is_enable")]
                mcp_logo = await UrlUtil.to_absolute_url(mcp.logo)
                mcp_srv[mcp.code] = {"name": mcp.name, "url": mcp.url, "tools": tool_list}
                self.mcp_tools[mcp.code] = {"name": mcp.name, "logo": mcp_logo}
            if mcp_srv:
                self.mcp_servers["mcpServers"] = mcp_srv

        # 查询用户
        if not self.publish_id:
            self.user = await UserModel.filter(id=self.user_id).first()
            if not self.user:
                raise AppException(msg="用户身份异常")
            if self.user.is_disable:
                raise AppException("用户已被拉黑")
        else:
            self.user = await UserModel.filter(id=self.robot.user_id).first()
            if not self.user:
                raise AppException("渠道发布者身份异常")
            if self.user.is_disable:
                raise AppException("用渠道发布者已被拉黑")

        # 余额验证
        if self.user.points <= 0:
            error = "账户余额不足,请充值后重试!"
            if self.search_points and self.search:
                raise AppException(error)
            if self.chat_price:
                raise AppException(error)
            if self.know_ids:
                if self.vector_price:
                    raise AppException(error)
                if self.ranking_price and self.robot.ranking_status:
                    raise AppException(error)

        # 违规审查
        review = CensorDriver.REVIEW_QUESTION
        censor = await CensorDriver.text(user_id=self.user.id, text=self.question, scene=review)
        if censor is not None:
            raise AppException(f"提问问题违规: {censor}")

    async def _get_apikey(self):
        """ 获取密钥 """
        apikey = {}
        if self.chat_config["check_key"]:
            apikey = await KeyPoolCache(scene="chat", channel=self.chat_mark).get_key() or {}
            if not apikey:
                raise AppException("当前模型尚未配置密钥,请与管理员联系")
        return apikey

    async def _get_instruct(self):
        """ 获取指令 """
        instruct = await (AgentRobotInstructModel
                          .filter(scene="fixed")
                          .filter(robot_id=self.robot.id)
                          .filter(keyword=self.question)
                          .order_by("-sort", "-id")
                          .first()
                          .values("id", "keyword", "content", "annex"))
        if not instruct:
            return []

        audio = []
        video = []
        files = []
        images = []
        scene = {"image": images, "audio": audio, "video": video, "files": files}
        annex = json.loads(instruct["annex"] or "[]")
        for item in annex:
            item["file_path"] = await UrlUtil.to_absolute_url(item["file_path"])
            array = scene[item["type"]]
            array.append({
                "file_type": str(item["file_type"]),
                "file_name": str(item["file_name"]),
                "file_size": int(item["file_size"]),
                "file_ext": str(item["ext"]),
                "file_path": await UrlUtil.to_absolute_url(str(item["file_path"]))
            })

        instruct["annex"] = images + video + audio + files
        return instruct

    async def _get_context(self):
        """ 取上下文 """
        attach: List[Dict[str, any]] = []    # 附带的文件
        systems: List[Dict[str, any]] = []   # 系统提示词
        messages: List[Dict[str, any]] = []  # 对话提示词

        # 获取历史记录
        history: list = await self._get_history()

        # 获取知识库内容
        search_result = ""
        if self.web_search:
            search_result = json.dumps({
                "type": "web_search",
                "tips": "联网搜索结果",
                "result": self.web_search
            }, indent=4, ensure_ascii=False)

        # 循环处理上下文
        file_content: str = ""
        chat_context: list = history + self.user_inputs
        attachments: dict = await self._query_db_files(chat_context)
        for index, context in enumerate(chat_context):
            if not isinstance(context["content"], list):
                messages.append(context)
            else:
                # 处理附件
                unique_md5 = []
                msg_content = []
                for i, ctx in enumerate(context["content"]):
                    mine = ctx.get("type")
                    file = attachments.get(ctx.get("file_id"))
                    if mine == "text":
                        continue
                    if not file:
                        del ctx
                        continue

                    att = {
                        "type": mine,
                        "file_id": file["id"],
                        "file_name": file["file_name"],
                        "file_size": file["file_size"],
                        "file_ext": file["file_ext"],
                        "file_path": file["file_path"]
                    }

                    context["content"][i] = att
                    if ctx.get("scene") != "history":
                        attach.append(att)
                    if file["file_md5"] in unique_md5:
                        continue
                    else:
                        unique_md5.append(file["file_md5"])

                    if mine in ["image", "video", "audio"]:
                        key: str = f"{mine}_url"
                        url: str = await UrlUtil.to_absolute_url(file["file_path"])
                        msg_content.append({"type": key, key: {"url": url}})
                    elif mine == "file":
                        anx = copy.copy(file)
                        del anx["id"]
                        del anx["file_path"]
                        del anx["text_path"]
                        file_content += "\n\n" if file_content else ""
                        file_content += json.dumps(anx, indent=4, ensure_ascii=False)

                # 整合问题
                if msg_content:
                    msg_content.append({"type": "text", "text": self.question})
                    messages.append({"role": "user", "content": msg_content})
                else:
                    messages.append({"role": "user", "content": self.question})

        # 全局系统指令
        if self.chat_config.get("global_directives"):
            directives = self.chat_config.get("global_directives")
            systems.append({"role": "system", "content": directives})

        # 设置角色指令
        if self.robot.prompt_roles:
            systems.append({"role": "system", "content": self.robot.prompt_roles})

        # 智能体限制指令
        quote_content = self.document or "" + file_content + search_result
        prompt: str = self.robot.prompt_limit or AgentEnum.get_prompt_tpl("chat")
        prompts = prompt.replace("[[document]]", quote_content)
        systems.append({"role": "system", "content": prompts})

        # 记录全局上下文
        self.attach = attach
        self.contexts = chat_context
        self.messages = systems + messages
        return self.messages

    async def _get_history(self):
        """ 取历史对话 """
        messages = []
        llm_configs: dict = json.loads(self.robot.llm_config)
        context_num: int = int(llm_configs.get("context_num", 0))
        if context_num > 0:
            records = await (AgentChatRecordsModel
                             .filter(scene=self.scene)
                             .filter(user_id=self.user_id)
                             .filter(robot_id=self.robot.id)
                             .filter(session_id=self.sid if self.scene == "normal" else 0)
                             .filter(is_clear=0)
                             .filter(is_delete=0)
                             .filter(user_delete=0)
                             .limit(context_num)
                             .order_by("-id")
                             .all())

            records = list(reversed(records))
            for record in records:
                msg_content = record.question
                attach = json.loads(record.attach or "[]")
                if not record.question or not record.answer:
                    continue
                if len(attach) > 0:
                    msg_content = []
                    for item in attach:
                        if item.get("type"):
                            msg_content.append({
                                "scene": "history",
                                "type": item["type"],
                                "file_id": item["file_id"]
                            })
                    msg_content.append({"type": "text", "text": record.question})
                messages.append({"role": "user", "content": msg_content})
                messages.append({"role": "assistant", "content": record.answer})

        return messages

    async def _call_learning(self):
        """ 知识库搜索 """
        start_time = time.time()
        if not self.know_ids:
            return None

        # 问题优化
        questions = [self.question]
        if self.robot.optimize_ask:
            questions.extend(await self._optimize_ask())

        # 向量模型
        vector_model = f"{self.vector_mark}:{self.vector_model}"

        # 发起检索
        results = []
        for query in questions:
            # 按模式检索
            if self.robot.search_mode == "similar":
                lists = await RecallKnowLogic.embedding_recall(
                    model=vector_model,
                    query=query,
                    knows=self.know_ids,
                    limit=100
                )
                results.append({"k": 60, "list": lists})
            elif self.robot.search_mode == "full":
                lists = await RecallKnowLogic.full_text_recall(
                    query=query,
                    knows=self.know_ids,
                    limit=100
                )
                results.append({"k": 60, "list": lists})
            elif self.robot.search_mode == "mix":
                lists = results.extend(await RecallKnowLogic.mixed_recall(
                    model=vector_model,
                    query=query,
                    knows=self.know_ids
                ))
                results.append({"k": 60, "list": lists})

        # 检索费用
        if self.robot.search_mode in ["similar", "mix"]:
            use_content: str = "".join(questions)
            total_tokens: int = ChatUtils.compute_tokens(use_content)
            consume_points: int = ChatUtils.compute_price(total_tokens, self.vector_price)
            # 流程登记
            self.flows_usage.append(FlowsSchema(
                name="文档搜索",
                scene="vector",
                alias=self.vector_alias,
                model=self.vector_model,
                task_time=str(f"{(time.time() - start_time):.2f}"),
                use_points=consume_points,
                total_tokens=total_tokens,
                prompt_tokens=total_tokens,
                completion_tokens=0
            ))

        # 结果处理(1): RRF重排
        results = RecallUtil.rrf_concat_results(results)

        # 结果处理(2): 重排模型
        similar: Decimal = self.robot.search_similar
        if self.robot.ranking_status:
            models = self.robot.ranking_model
            similar = self.robot.ranking_score
            data = await RecallUtil.re_ranker_results(results, question=self.question, models=models)
            results = data["results"]
            self.rerank_usage = data["usage"]
            self.robot.ranking_status = data["status"]

        # 结果处理(3): 相似度过滤
        results = RecallUtil.filter_max_score(
            results=results,
            mode=self.robot.search_mode,
            rerank=self.robot.ranking_status,
            similar=similar
        )

        # 结果处理(4): MaxTokens过滤
        results = RecallUtil.filter_max_tokens(results=results, max_tokens=self.robot.search_tokens)

        # 结果处理(5): 整合引用知识
        index: int = 0
        uuids: list = []
        quotes: list = []
        content: str = ""
        for item in results:
            uuids.append(str(item["uuid"]))
            quotes.append(item)
            if index > 0:
                content += "\n------------\n\n"

            index += 1
            texts = item["question"] + "\n"
            if item.get("answer"):
                texts += item["answer"] + "\n"

            content += json.dumps({
                "source_name": item.get("file", {}).get("name", "-"),
                "weight_score": item.get("emb_score", 0),
                "content": texts
            }, ensure_ascii=False, indent=4)

        # 召回附件(6): 根据语义相似度
        if uuids:
            self.adjuncts = await RecallKnowLogic.annex_recall(
                model=vector_model,
                query=self.question,
                ids=uuids,
                similar=self.robot.annex_similar
            )

        # 返回结果结果
        self.quotes = quotes
        self.document = content
        return self.document

    async def _optimize_ask(self):
        """ 问题优化 """
        if not self.robot.optimize_ask:
            return []

        start_time = time.time()
        records = await (AgentChatRecordsModel
                         .filter(scene=self.scene)
                         .filter(user_id=self.user_id)
                         .filter(robot_id=self.robot.id)
                         .filter(session_id=self.sid if self.scene == "normal" else 0)
                         .filter(is_clear=0)
                         .filter(is_delete=0)
                         .limit(3)
                         .order_by("-id")
                         .all().values("id", "question", "answer"))

        history = ""
        for item in records:
            q = item["question"]
            a = item["answer"]
            if history:
                history += "\n"
            history += f"Q: {q}\nA: {a}"

        template = AgentEnum.get_prompt_tpl("optimize")
        template = template.replace("{{histories}}", history)
        template = template.replace("{{query}}", self.question)
        messages = [{"role": "user", "content": template}]

        # 验证优化模型
        try:
            models = await DevModelsModel.check_models(self.robot.optimize_model, "chat")
            mark = models["mark"]
            alias = models["alias"]
            model = models["model"]
            price = models["price"]
            config = models["config"]
            channel = models["channel"]
        except Exception as e:
            raise AppException(str(e))

        # 获取模型密钥
        apiKey = {}
        if config["check_key"]:
            apiKey = await KeyPoolCache(scene="chat", channel=mark).get_key() or {}
            if not apiKey:
                raise AppException("问题优化模型尚未配置密钥,请与管理员联系")

        # 发起对话请求
        aiServer = AIChannelFactory.create(channel, config, apiKey)
        results = await aiServer.llm(messages, stream=False)
        content = results.choices[0].message.content

        # 计算消耗费用
        consume_points: int = ChatUtils.compute_price(results.usage.total_tokens, int(price))
        consume_points: int = 0 if price else consume_points
        self.flows_usage.append(FlowsSchema(
            name="问题优化",
            scene="optimize",
            alias=alias,
            model=model,
            task_time=f"{(time.time() - start_time):.2f}",
            use_points=consume_points,
            total_tokens=results.usage.total_tokens,
            prompt_tokens=results.usage.prompt_tokens,
            completion_tokens=results.usage.completion_tokens
        ))

        # 返回问题列表
        return json.loads(content.strip('```json ').strip('```').strip())

    async def _web_searching(self):
        """ 联网搜索 """
        if self.search and self.search_status:
            start_time = time.time()
            self.web_search = await AIWebSearch.search(query=self.question)
            self.flows_usage.append(FlowsSchema(
                name="联网搜索",
                scene="web_search",
                alias="智谱搜索",
                model=self.search_engine,
                task_time=f"{(time.time() - start_time):.2f}",
                use_points=self.search_points,
                total_tokens=0,
                prompt_tokens=0,
                completion_tokens=0,
            ))

    async def _make_related(self) -> None:
        """ 生成相关问题 """
        try:
            template = AgentEnum.get_prompt_tpl(scene="issues")
            messages = [
                *self.contexts,
                {"role": "assistant", "content": self.reply},
                {"role": "user", "content": template},
            ]

            apikey = await self._get_apikey()
            aiServer = AIChannelFactory.create(self.chat_channel, self.chat_config, apikey)
            results = await aiServer.llm(messages, stream=False)
            content = results.choices[0].message.content
            arrays = json.loads(content.strip('```json ').strip('```').strip())
            issues = json.dumps(arrays, ensure_ascii=False)
            await AgentChatRecordsModel.filter(id=self.record_id).update(issues=issues)
        except Exception as e:
            print(e)

    async def _save_record(self) -> None:
        """ 保存记录 """
        async with in_transaction("mysql"):
            # 重查用户
            user_id: int = self.robot.user_id if self.publish_id else self.user_id
            self.user = await UserModel.filter(id=user_id).first()

            # 累计消耗费用
            total_consume_tokens: int = sum(flow.total_tokens for flow in self.flows_usage)
            total_consume_points: int = sum(flow.use_points for flow in self.flows_usage)

            # 用户扣减费用
            if total_consume_points:
                surplus_balance: int = max(self.user.points - total_consume_points, 0)
                await UserModel.filter(id=user_id).update(points=surplus_balance)

            # 写入记录
            flows_usage = [flow.model_dump() for flow in self.flows_usage]
            record = await AgentChatRecordsModel.create(
                scene=self.scene,
                user_id=user_id,
                robot_id=self.robot.id,
                session_id=self.sid,
                publish_id=self.publish_id,
                publish_uid=self.publish_uid,
                use_tokens=total_consume_tokens,
                use_points=total_consume_points,
                question=self.question,
                answer=self.reply,
                thinks=self.think,
                quotes=json.dumps(self.quotes, ensure_ascii=False),
                context=json.dumps(self.contexts, ensure_ascii=False),
                attach=json.dumps(self.attach, ensure_ascii=False),
                annex=json.dumps(self.annex, ensure_ascii=False),
                tools_list=json.dumps(self.tools_list, ensure_ascii=False),
                web_search=json.dumps(self.web_search, ensure_ascii=False),
                flows_usage=json.dumps(flows_usage, ensure_ascii=False),
                ip=self.ip,
                task_time=f"{(time.time() - self.start_time):.2f}",
                create_time=int(time.time()),
                update_time=int(time.time())
            )

            # 记录的ID
            self.record_id = record.id

            # 扣费日志
            if total_consume_points:
                await UserWalletModel.dec(
                    scene=WalletEnum.POINTS,
                    user_id=user_id,
                    change_type=WalletEnum.POINTS_DEC_AGENT_CHAT,
                    change_amount=total_consume_points,
                    remarks=WalletEnum.get_change_type_msg(WalletEnum.POINTS_DEC_AGENT_CHAT),
                    project=self.robot.name,
                    source_sn=self.robot.code,
                    source_id=self.robot.id,
                    additional=json.dumps(flows_usage, ensure_ascii=False)
                )

            # 发布更新
            if self.publish_id:
                publish = await AgentRobotPublishModel.filter(id=self.publish_id).first()
                await AgentRobotPublishModel.filter(id=self.publish_id).update(
                    use_points=publish.use_points + total_consume_points,
                    call_count=publish.call_count + 1,
                    last_ip=self.ip,
                    last_time=int(time.time())
                )

    async def _query_db_files(self, context: list) -> dict:
        """ 从数据库查附件 """
        attrs: dict = {}
        file_ids: list = []

        # 提取文件的ID
        for index, msgs in enumerate(context):
            if isinstance(msgs["content"], list):
                file_ids = [item["file_id"] for item in msgs["content"] if item.get("file_id")]

        # 查找文件数据
        if file_ids:
            fields = "id,file_md5,file_type,file_name,file_size,file_ext,file_path,text_path".split(",")
            attachments = await AgentAttachmentModel.filter(id__in=set(file_ids)).values(*fields)
            attrs = {item["id"]: item for item in attachments}

        # 读取文件内容
        for _, item in attrs.items():
            item["file_text"] = ""
            if item["file_type"] == "file":
                content = await self._read_file_content(item["file_path"], item["text_path"])
                item["file_text"] = content.strip()

        # 返回查询数据
        return attrs

    @classmethod
    async def _read_file_content(cls, file_path: str, text_path: str) -> str:
        """
        读取对话文件内容

        Args:
            file_path (str): 文件路径: (如: storage/agents/chat/20250414/aa.docx)
            text_path (str): 文本路径: (如: parsing/chats/20250414/aa.md)

        Returns:
            str: 文件的内容

        Author:
            zero
        """
        if not file_path:
            return ""

        # 读取内容
        md_name = ToolsUtil.make_md5_str(file_path)
        md_path = UrlUtil.to_runtime_path(text_path)
        if os.path.exists(md_path):
            return await ToolsUtil.read_file(md_path)
        # 重新读取
        root_file = UrlUtil.to_root_path(file_path)
        if os.path.exists(root_file):
            # 文件存在本地: 直接重新读取
            content = await FilesUtil.parse_raw(root_file)
            await ToolsUtil.write_file(md_path, content)
            return content
        else:
            # 文件不在本地: 远程下载读取
            file_ext = file_path.split(":")[-1]
            temp_path = UrlUtil.to_runtime_path(f"temporary/{md_name}.{file_ext}")
            try:
                # 可能是OSS(尝试远程下载)
                await ToolsUtil.download_file(await UrlUtil.to_absolute_url(file_path), temp_path)
                # 读取下载的文件
                content = await FilesUtil.parse_raw(root_file)
                # 保存读取的内容
                await ToolsUtil.write_file(md_path, content)
                return content
            except Exception as e:
                print(str(e))
                return ""
            finally:
                if os.path.exists(temp_path):
                    os.remove(temp_path)
